package com.atguigu.source;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.connector.source.util.ratelimit.RateLimiterStrategy;
import org.apache.flink.connector.datagen.source.DataGeneratorSource;
import org.apache.flink.connector.datagen.source.GeneratorFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author gmd
 * @desc 数据生成器
 * @since 2024-11-24 10:10:38
 */
public class DataGeneratorData {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 并行度设置
        // 如果有n个并行度， 最大值设为a
        // 将数值 均分成 n份，  a/n ,比如，最大100，并行度2，每个并行度生成50个
        // 其中一个是 0-49，另一个50-99
        // 详细解释：当并行度大于1时，会将数据均分成多个部分，每个部分由一个并行度处理
        //          例如，如果最大值为100，并行度为2，则每个并行度处理50个数据
        //          第一个并行度处理0-49，第二个并行度处理50-99
        env.setParallelism(3);

        /*
         * 数据生成器Source，四个参数：
         *     第一个参数： GeneratorFunction接口，需要实现，重写map方法，输入类型固定是Long
         *          GeneratorFunction是一个接口，需要实现map方法，用于将输入的Long类型转换为所需的输出类型
         *     第二个参数： long类型，自动生成的数字序列（从0自增）的最大值(小于)，达到这个值就停止了
         *          DataGeneratorSource会生成一个自增的Long类型数字序列，当达到这个最大值时，生成器停止
         *     第三个参数： 限速策略，比如每秒生成几条数据
         *          RateLimiterStrategy是一个枚举类，提供了不同的限速策略，比如每秒生成几条数据
         *     第四个参数： 返回的类型
         *          Types是一个类，提供了不同类型的类型信息，用于指定DataGeneratorSource的返回类型
         */
        DataGeneratorSource<String> dataGeneratorSource = new DataGeneratorSource<>(
                new GeneratorFunction<Long, String>() {
                    @Override
                    public String map(Long value) throws Exception {
                        return "Number:" + value;
                    }
                },
                100,
                RateLimiterStrategy.perSecond(5),
                Types.STRING
        );

        env.fromSource(dataGeneratorSource, WatermarkStrategy.noWatermarks(), "data-generator")
                .print();

        env.execute();
    }

}
